package com.intct.flink.study;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-13 15:25
 * kafka source -> kafka sink
 */
public class SQLTest13 {
    public static void main(String[] args) {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8081");

        // 1 获取取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 并行度
         env.setParallelism(1);
        // 开启检点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 获取SQL执行环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 映射mysql的订单
        tenv.executeSql(
                "create table orders( " +
                        // 如果是Flink sql的关键字，需要用反引号转义
                        "  `user`        BIGINT " +
                        "  ,product        STRING " +
                        "  ,order_time    TIMESTAMP(3) " +
                        "  ,customer_id    INT " +
                        "  ,order_id    INT " +
                        // 使用proctime获取数据处理时间
                        "  ,proc_time as proctime() " +
                        // 水位线
                        "  ,watermark for order_time as order_time - interval '0' second " +
                        // 设置主键
                        "    ,PRIMARY KEY (`user`) NOT ENFORCED " +
                        ") with ( " +
                        "  'connector' = 'mysql-cdc' " +
                        "  ,'hostname' = 'cdh-node' " +
                        "  ,'port' = '13306' " +
                        "  ,'username' = 'root' " +
                        "  ,'password' = 'Test_090110' " +
                        "  ,'database-name' = 'm1' " +
                        "  ,'table-name' = 'orders' " +
                        ") ");

        tenv.executeSql("create table redis_table(       " +
                "  key_name STRING                                " +
                "  ,field STRING                                  " +
                "  ,`value` STRING                                " +
                ") WITH (                                         " +
                "  'connector'='redis'                            " +
                "  ,'host'='cdh-node'                             " +
                "  ,'port'='16379'                                " +
                "  ,'redis-mode'='single'                         " +
                "  ,'password'='Test_090110'                      " +
                "  ,'ttl'='60'                      " +
                "  ,'command'='hset'                              " +
                ")                                                ");

        tenv.executeSql("insert into redis_table select cast(order_id as string), product, cast(`user` as string) from orders");
    }
}
